package com.cn.common;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class ProcessQueue {

    //最大锁定时间
    public final static long REBALANCE_LOCK_MAX_LIVE_TIME = 30000;

    public final static long REBALANCE_LOCK_INTERVAL = 20000;

    public final static long PULL_MAX_IDLE_TIME = 120000;

    //读写锁
    private final ReadWriteLock lockTreeMap = new ReentrantReadWriteLock();
    private final TreeMap<Long, MessageExt> treeMap = new TreeMap<Long, MessageExt>();

    //消息总数
    private final AtomicLong msgCount = new AtomicLong(0);
    //消息大小
    private final AtomicLong msgSize = new AtomicLong(0);

    private final Lock lockConsume = new ReentrantLock();
    private final TreeMap<Long, MessageExt> consumingMsgOrderlyTreeMap = new TreeMap<Long, MessageExt>();

    //尝试释放锁的总次数
    private final AtomicLong tryUnlockTimes = new AtomicLong();

    //当前队列最大的消费位移
    private volatile long queueOffsetMax = 0;
    //该进程是否可用
    private volatile boolean dropped = false;


    //最后一次拉取消息的时间
    private volatile long lastPullTimestamp = System.currentTimeMillis();
    //最后一次消费时间
    private volatile long lastConsumeTimestamp = System.currentTimeMillis();
    //该进程是否已经在broker端加锁
    private volatile boolean locked = false;
    //最后一次锁定时间
    private volatile long lastLockTimestamp = System.currentTimeMillis();

    private volatile long nextOffset = 0;

    //是否正在消费
    private volatile boolean consuming = false;

    private volatile long msgAccCnt = 0;

    private volatile boolean lockFirst = false;


    //斥锁是否过期
    public boolean isLockExpired() {
        return (System.currentTimeMillis() - this.lastLockTimestamp) > REBALANCE_LOCK_MAX_LIVE_TIME;
    }

    //拉取消息是否超时
    public boolean isPullExpired() {
        return (System.currentTimeMillis() - this.lastPullTimestamp) > PULL_MAX_IDLE_TIME;
    }


    public ReadWriteLock getLockTreeMap() {
        return lockTreeMap;
    }

    public AtomicLong getMsgCount() {
        return msgCount;
    }

    public AtomicLong getMsgSize() {
        return msgSize;
    }

    public Lock getLockConsume() {
        return lockConsume;
    }

    public AtomicLong getTryUnlockTimes() {
        return tryUnlockTimes;
    }

    public long getQueueOffsetMax() {
        return queueOffsetMax;
    }

    public boolean isDropped() {
        return dropped;
    }

    public long getLastPullTimestamp() {
        return lastPullTimestamp;
    }

    public long getLastConsumeTimestamp() {
        return lastConsumeTimestamp;
    }

    public boolean isLocked() {
        return locked;
    }

    public long getLastLockTimestamp() {
        return lastLockTimestamp;
    }

    public boolean isConsuming() {
        return consuming;
    }

    public long getMsgAccCnt() {
        return msgAccCnt;
    }

    public void setDropped(boolean dropped) {
        this.dropped = dropped;
    }

    public void setLocked(boolean locked) {
        this.locked = locked;
    }

    public void setLastPullTimestamp(long lastPullTimestamp) {
        this.lastPullTimestamp = lastPullTimestamp;
    }

    public void setLastConsumeTimestamp(long lastConsumeTimestamp) {
        this.lastConsumeTimestamp = lastConsumeTimestamp;
    }

    public void setLastLockTimestamp(long lastLockTimestamp) {
        this.lastLockTimestamp = lastLockTimestamp;
    }

    public void setConsuming(boolean consuming) {
        this.consuming = consuming;
    }

    /**
     * 存放等待消费的消息
     *
     * @param msgs
     */
    public void putMessage(List<MessageExt> msgs) {
        if (msgs == null) return;
        try {
            this.lockTreeMap.writeLock().lockInterruptibly();
            try {
                for (MessageExt msg : msgs) {
                    MessageExt old = this.treeMap.putIfAbsent(msg.getQueueoffset(), msg);
                    if (null == old) {
                        //当前进程中的消息总数
                        this.msgCount.incrementAndGet();
                        //当前进程中累计的消息总量
                        this.msgSize.addAndGet(msg.getBody().length);
                    }
                }
            } finally {
                this.lockTreeMap.writeLock().unlock();
            }
        } catch (InterruptedException e) {

        } finally {

        }
    }

    /**
     * 消费消息
     *
     * @param size
     * @return
     */
    public List<MessageExt> takeMessage(int size) {
        List<MessageExt> msgs = new ArrayList<>(size);
        try {
            try {
                this.lockTreeMap.writeLock().lockInterruptibly();
                for (int i = 0; i < size; i++) {
                    Map.Entry<Long, MessageExt> msg = this.treeMap.pollFirstEntry();
                    if (msg == null) {
                        break;
                    }
                    msgs.add(msg.getValue());
                    //加入到消费队列中
                    this.consumingMsgOrderlyTreeMap.putIfAbsent(msg.getKey(), msg.getValue());
                }
            } finally {
                this.lockTreeMap.writeLock().unlock();
            }

        } catch (InterruptedException e) {

        }
        return msgs;
    }

    /**
     * 提交，并返回消费位移
     *
     * @return
     */
    public Long commit() {
        try {
            this.lockTreeMap.writeLock().lockInterruptibly();
            try {
                Long offset = this.consumingMsgOrderlyTreeMap.lastKey();

                this.msgCount.addAndGet(0 - this.consumingMsgOrderlyTreeMap.size());
                for (MessageExt messageExt : this.consumingMsgOrderlyTreeMap.values()) {
                    this.msgSize.addAndGet(0 - (messageExt.getBody().length));
                }
                if (offset != null) {
                    offset = offset + 1;
                    return offset;
                }
            } finally {
                this.lockTreeMap.writeLock().unlock();
            }
        } catch (InterruptedException e) {

        }
        return -1l;
    }

    /**
     * 回滚
     *
     * @param msg
     */
    public void rollback(List<MessageExt> msg) {
        try {
            this.lockTreeMap.writeLock().lockInterruptibly();
            try {
                this.treeMap.putAll(this.consumingMsgOrderlyTreeMap);
                this.consumingMsgOrderlyTreeMap.clear();
            } finally {
                this.lockTreeMap.writeLock().unlock();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 是否存在等待处理的消息
     *
     * @return
     */
    public boolean hasTempMsg() {
        try {
            this.lockTreeMap.readLock().lockInterruptibly();
            try {
                return !this.treeMap.isEmpty();
            } finally {
                this.lockTreeMap.readLock().unlock();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return true;
    }

    public boolean isLockFirst() {
        return lockFirst;
    }

    public void setLockFirst(boolean lockFirst) {
        this.lockFirst = lockFirst;
    }

    public long getNextOffset() {
        return nextOffset;
    }

    public void setNextOffset(long nextOffset) {
        this.nextOffset = nextOffset;
    }

    public static void main(String[] args) {
        AtomicLong l = new AtomicLong(0);
        System.out.println("1->" + l.incrementAndGet());
    }

}
